package cn.lecosa.spark.mysql
import java.util.Properties
import org.apache.spark.sql.{ SQLContext, Row }
import org.apache.spark.sql.types.{ StringType, IntegerType, StructField, StructType }
import org.apache.spark.{ SparkConf, SparkContext }
import org.apache.spark.sql.SparkSession

object Df2Mysql {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("MySQL-Demo")
    conf.setAppName("jdbcApp!"); //设置应用程序的名称，在spark程序运行的监控界面可以看到名称
    conf.setMaster("local[2]"); //此时，程序在本地运行，不需要安装spark集群
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    //通过并行化创建RDD
    val personRDD = sc.parallelize(Array("tom", "jerry", "kitty")).map(_.split(" "))

    //通过StructType直接指定每个字段的schema
    val schema = StructType(
      List(
        //        StructField("id", IntegerType, true),
        StructField("name", StringType, true) //        StructField("age", IntegerType, true)
        ))
    //将RDD映射到rowRDD
    //    val rowRDD = personRDD.map(p => Row(p(0).toInt, p(1).trim, p(2).toInt))
    val rowRDD = personRDD.map(p => Row(p(0).trim))
    //将schema信息应用到rowRDD上
    val personDataFrame = sqlContext.createDataFrame(rowRDD, schema)
    //创建Properties存储数据库相关属性
    val prop = new Properties()
    prop.put("user", "root")
    prop.put("password", "root")
    //将数据追加到数据库
    //    personDataFrame.write.mode("append").jdbc("jdbc:mysql://192.168.164.141:3306/jtdb", "jtdb.test", prop)

    val df = sqlContext.read.format("jdbc").options(Map("url" -> "jdbc:mysql://10.20.8.95:3306/pip", "driver" -> "com.mysql.jdbc.Driver", "dbtable" -> "pip.t_device", "user" -> "root", "password" -> "skycomm@123")).load()
    personDataFrame.show();
    //停止SparkContext
    sc.stop()

  }
}